package common

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/mvccpb"
	etcd "go.etcd.io/etcd/client/v3"
)

func newClient(host string, cancel context.CancelFunc) (*etcd.Client, error) {
	cfg := etcd.Config{
		Endpoints:   []string{host},
		DialTimeout: 5 * time.Second,
	}

	client, err := etcd.New(cfg)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("连接etcd失败: %w", err)
	}
	return client, nil
}

type Registry struct {
	client     *etcd.Client
	keep_alive <-chan *etcd.LeaseKeepAliveResponse
	leaseID    etcd.LeaseID
	ctx        context.Context
	cancel     context.CancelFunc
	sync.Mutex
	close bool
}

func NewRegistry(host string) (*Registry, error) {
	ctx, cancel := context.WithCancel(context.Background())

	client, err := newClient(host, cancel)
	if err != nil {
		return nil, err
	}

	leaseResp, err := client.Lease.Grant(ctx, 3)
	if err != nil {
		client.Close()
		cancel()
		return nil, fmt.Errorf("创建租约失败: %w", err)
	}
	leaseID := leaseResp.ID

	keep_alive, err := client.Lease.KeepAlive(ctx, leaseID)
	if err != nil {
		client.Close()
		cancel()
		return nil, fmt.Errorf("启动租约保活失败: %w", err)
	}

	go func() {
		for {
			select {
			case resp, ok := <-keep_alive:
				if !ok || resp == nil {
					log.Println("保活通道已关闭")
					return
				}
			case <-ctx.Done():
				log.Println("保活协程终止")
				return
			}
		}
	}()

	return &Registry{
		client:     client,
		leaseID:    leaseID,
		keep_alive: keep_alive,
		ctx:        ctx,
		cancel:     cancel,
		close:      false,
	}, nil
}

func (r *Registry) Close() {
	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()

	if r.close {
		return
	}

	if r.cancel != nil {
		r.cancel()
	}

	if r.client != nil {
		_ = r.client.Close()
	}

	r.close = true
}

func (r *Registry) Register(key string, val string) error {
	r.Lock()
	defer r.Unlock()

	if r.close {
		log.Println("Registry 已关闭,禁止操作")
		return nil
	}

	ctx, cancel := context.WithTimeout(r.ctx, 3*time.Second)
	defer cancel()

	_, err := r.client.Put(ctx, key, val, etcd.WithLease(r.leaseID))
	if err != nil {
		return fmt.Errorf("服务注册失败: key: %s,value: %s %w", key, val, err)
	}

	return nil
}

type NotifyCallback func(key, value string)

type Discovery struct {
	client    *etcd.Client
	baseDir   string
	putCb     NotifyCallback
	delCb     NotifyCallback
	watchChan <-chan etcd.WatchResponse
	ctx       context.Context
	cancel    context.CancelFunc
	sync.Mutex
	close bool
}

func NewDiscovery(host, baseDir string, putCb, delCb NotifyCallback) (*Discovery, error) {
	ctx, cancel := context.WithCancel(context.Background())

	client, err := newClient(host, cancel)
	if err != nil {
		return nil, err
	}

	disc := &Discovery{
		client:  client,
		baseDir: baseDir,
		putCb:   putCb,
		delCb:   delCb,
		ctx:     ctx,
		cancel:  cancel,
		close:   false,
	}

	if err := disc.initDiscover(); err != nil {
		disc.Close()
		return nil, fmt.Errorf("服务发现初始化失败: %w", err)
	}

	return disc, nil
}

func (d *Discovery) Watch() {
	go d.watching()
}

func (d *Discovery) initDiscover() error {
	resp, err := d.client.Get(d.ctx, d.baseDir, etcd.WithPrefix())
	if err != nil {
		return fmt.Errorf("获取服务列表失败: %w", err)
	}

	for _, kv := range resp.Kvs {
		key := string(kv.Key)
		val := string(kv.Value)
		log.Printf("{%s} 提供 {%s} 服务", key, val)
		if d.putCb != nil {
			d.putCb(key, val)
		}
	}
	return nil
}

func (d *Discovery) watching() {
	// 启动携带后才能获取删除前的value
	watchChan := d.client.Watch(d.ctx, d.baseDir, etcd.WithPrefix(), etcd.WithPrevKV())
	if watchChan == nil {
		log.Fatal("启动监听失败")
		return
	}
	d.watchChan = watchChan

	for watchResp := range watchChan {
		for _, event := range watchResp.Events {
			switch event.Type {
			case etcd.EventTypePut:
				key := string(event.Kv.Key)
				val := string(event.Kv.Value)
				log.Printf("新增服务 {%s}-{%s}", key, val)
				if d.putCb != nil {
					d.putCb(key, val)
				}
			case mvccpb.DELETE:
				key := string(event.PrevKv.Key)
				val := string(event.PrevKv.Value)
				log.Printf("服务 {%s}-{%s} 下线", key, val)
				if d.delCb != nil {
					d.delCb(key, val)
				}
			}
		}
	}
}

func (d *Discovery) Close() {
	d.Lock()
	defer d.Unlock()

	if d.close {
		return
	}

	d.close = true
	d.cancel()
	d.client.Close()
}
